/*
 * Hyuga: A Density-Grid Stream Clustering Platform.
 * Copyright (C) 2014 PUC-Rio/Laboratory for Advanced Collaboration
 *
 * Hyuga is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Hyuga is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Hyuga.  If not, see <http://www.gnu.org/licenses/>.
 */
package br.pucrio.inf.lac.konoha.hyuga.core;

import java.util.Deque;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

import org.apache.log4j.Logger;

import br.pucrio.inf.lac.konoha.hyuga.common.Pair;
import br.pucrio.inf.lac.konoha.hyuga.core.output.DG2CEPEntry;
import br.pucrio.inf.lac.konoha.hyuga.core.output.OutputEntry;
import br.pucrio.inf.lac.konoha.hyuga.core.output.OutputType;
import br.pucrio.inf.lac.konoha.hyuga.events.LocationUpdateEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.cell.CellContentEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.semantics.DispersedGridClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.semantics.MergeGridClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.semantics.NewCellClusterEvent;
import br.pucrio.inf.lac.konoha.hyuga.events.semantics.UpdateGridClusterEvent;

/**
 * A standard output end point for the Hyuga platform.
 * 
 * This class register itself as listener for several semantic cluster output of Hyuga EPN, such as cluster formation,
 * merge, dispersion, etc. The events are stored in the Queue for further analysis or can be pushed for consumer 
 * applications. 
 * 
 * @author Marcos Roriz <mroriz@inf.puc-rio.br>
 */
public class Output {
    /** LOGGER. */
    private static Logger LOGGER = Logger.getLogger(Output.class.getCanonicalName());

    /** Singleton Instance. */
    protected static Output instance = new Output();
    
    /** Output Queue. Useful for logging. */
    protected Deque<OutputEntry> outputQueue;
    
    /** Output Thread Pool. */
    private ExecutorService outPool;

    // Timestamp Variables
    protected boolean started;
    protected long initialTimestamp;
    
    // Singleton Constructor
    private Output() {
        this.started     = false;
        this.outPool     = Executors.newFixedThreadPool(4);
        this.outputQueue = new LinkedBlockingDeque<OutputEntry>();
    }
    
    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Get and Sets
    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    
    // Singleton
    public static Output getInstance() {
        return instance;
    }
    
    // Output Queue
    public Deque<OutputEntry> getOutputQueue() {
        return outputQueue;
    }
    
    public void clear() {
        this.outputQueue.clear();
    }
    
    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Timestamp function
    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    
    // Timestamp control
    public boolean hasStarted() {
        return started;
    }
    
    public void setup() {
        started = true;
        this.initialTimestamp = System.currentTimeMillis();
    }

    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Update listeners for Semantic Operations
    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    
    // Output Action
    public class OutputAction implements Runnable {
        private Object event;
        private CellContentEvent[] cellContents;
        private OutputType type;

        public OutputAction(Object event, CellContentEvent[] cellContents, OutputType type) {
            this.event = event;
            this.cellContents = cellContents;
            this.type = type;
        }
        
        @Override
        public void run() {
            DG2CEPEntry entry = new DG2CEPEntry();
            // Basic
            entry.setType(type);
            entry.setTimestamp(System.currentTimeMillis() - initialTimestamp);
            
            // Content (LU)
            entry.setSize(size(cellContents));                        
            entry.setCellContents(cellContents);
            entry.setStrContent(stringifyClusterContent(cellContents));

                        
            Set<Pair<Integer, Integer>> coreCells = null;
            switch (type) {
                case ADD:
                    NewCellClusterEvent addCellCluster = (NewCellClusterEvent) event;
                    coreCells = addCellCluster.getCluster().getCoreCells();
                    break;
                case UPDATE:
                    UpdateGridClusterEvent upCellCluster = (UpdateGridClusterEvent) event;
                    coreCells = upCellCluster.getCoreCells();
                    break;
                case MERGE:
                    MergeGridClusterEvent mergeCluster = (MergeGridClusterEvent) event;
                    coreCells = mergeCluster.getCoreCells();
                    break;
                case DISPERSE:
                    DispersedGridClusterEvent disperseEvent = (DispersedGridClusterEvent) event;
                    coreCells = disperseEvent.getCluster().getCoreCells();
                    entry.setSize(-1);
                    entry.setStrContent("");
                default:
                    break;
            }
            
            // Cells
            entry.setClusterCells(coreCells);
            entry.setStrClusterCells(stringifyClusterCells(coreCells));

            synchronized (outputQueue) {
                outputQueue.add(entry);
            }
            LOGGER.info("=== " + type + " " + event + " - SIZE: " + size(cellContents));
        }
        
    }
    
    
    // ADD
    public void update(NewCellClusterEvent addCellCluster) {
        OutputAction out = new OutputAction(addCellCluster, addCellCluster.getCell().toCellContentArray(), OutputType.ADD);
        outPool.submit(out);
    }
    
    // UPDATE
    public void update(UpdateGridClusterEvent upCellCluster, CellContentEvent[] cellContents) {
        OutputAction out = new OutputAction(upCellCluster, cellContents, OutputType.UPDATE);
        outPool.submit(out);
    }
    
    // MERGE
    public void update(MergeGridClusterEvent mergeCluster, CellContentEvent[] cellContents) {
        if (mergeCluster.getCoreCells().size() > 1) {
            OutputAction out = new OutputAction(mergeCluster, cellContents, OutputType.MERGE);
            outPool.submit(out);
        }
    }
    
    // DISPERSE
    public void update(DispersedGridClusterEvent disperseEvent) {
        OutputAction out = new OutputAction(disperseEvent, new CellContentEvent[]{}, OutputType.DISPERSE);
        outPool.submit(out);
    }

    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // Helper Functions (Stringify, size, etc)
    //////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    public int size(CellContentEvent[] contentArray) {
        int size = 0;
        if (contentArray != null && contentArray.length > 0) { 
            for (CellContentEvent content : contentArray) {
                if (content != null) size += content.size(); 
            }
        }
        return size;
    }
    
    String stringifyClusterCells(Set<Pair<Integer, Integer>> coreCells) {
        String delim = "[";
        String clusterCells = "";
        for (Pair<Integer, Integer> cell : coreCells) {
            clusterCells += delim + "(" + cell.getFirst() + "-" + cell.getSecond() + ")";
            delim = "~";
        }
        return clusterCells + "]";
    }
    
    String stringifyClusterContent(CellContentEvent[] cellContents) {
        String delim = "";
        String clusterContent = "\"";
        if (cellContents != null) {
            for (CellContentEvent cell : cellContents) {
                if (cell != null) {
                    for (LocationUpdateEvent lu : cell.getNodes()) {
                        clusterContent += delim + lu.getOriginalID();
                        delim = ",";
                    }
                }
            }
        }
        clusterContent += "\"";
        return clusterContent;
    }
}
